package priv.mill.bigdata.hdfs;

import static org.junit.Assert.*;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.URI;
import java.nio.file.FileVisitOption;
import java.nio.file.Files;
import java.nio.file.OpenOption;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class HdfsClient {

	private FileSystem fs;
	private Configuration conf;

	@Before
	public void before() {
		conf = new Configuration();
		// conf.set("dfs.replication", "5");//最终这里的配置生效
		try {
			System.setProperty("HADOOP_USER_NAME", "root");
			fs = FileSystem.get(URI.create("hdfs://mini1:9000"), conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Test
	public void test1() throws Exception {
		Path src = new Path("/Users/mill/Documents/gits/boot/haddop-example/datas/auto-mpg.data-original");
		Path dst = new Path("/dataset/auto-mpg.data-original");
		fs.copyFromLocalFile(src, dst);
	}

	@Test
	public void test2() throws Exception {
		fs.copyToLocalFile(new Path("/aaa.mp4"), new Path("/Users/mill"));
	}

	/**
	 * 打印所有配置项
	 * 
	 * @throws Exception
	 */
	@Test
	public void printConfig() throws Exception {
		Iterator<Entry<String, String>> iterator = conf.iterator();
		while (iterator.hasNext()) {
			Entry<String, String> next = iterator.next();
			System.out.printf("name:%s;value:%s;\n", next.getKey(), next.getValue());
		}
	}

	/**
	 * 创建目录
	 * 
	 * @throws Exception
	 */
	@Test
	public void mkdir() throws Exception {
		boolean mkdirs = fs.mkdirs(new Path("/mktest/aaa"));
		if (mkdirs)
			System.out.println("创建目录成功");
	}

	/**
	 * 删除
	 * 
	 * @param path
	 *            删除路径
	 * @param r
	 *            是否递归删除
	 * @throws Exception
	 */
	@Test
	public void rmdir() throws Exception {
		boolean result = fs.delete(new Path("/mktest/aaa"), true);
		if (result)
			System.out.println("删除目录成功");
	}

	/**
	 * 列出所有文件(递归返回迭代器)
	 * 
	 * @throws Exception
	 */
	@Test
	public void list() throws Exception {
		RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
		while (listFiles.hasNext()) {
			LocatedFileStatus next = listFiles.next();
			Path path = next.getPath();
			System.out.println(path.getName());
			System.out.println(next.getBlockSize());
			System.out.println("====================");
			for (BlockLocation block : next.getBlockLocations()) {
				String[] hosts = block.getHosts();
				System.out.println("===========:所在主机"+Arrays.toString(hosts));
				System.out.println("===========:长度"+block.getLength());
				
			}
		}
	}

	/**
	 * 列出所有文件（不递归返回数组）
	 * 
	 * @throws Exception
	 */
	@Test
	public void list2() throws Exception {
		FileStatus[] listStatus = fs.listStatus(new Path("/"));
		for (FileStatus fileStatus : listStatus) {
			if (fileStatus.isDirectory()) {
				System.out.println("文件夹" + fileStatus.getPath().getName());
			} else {
				System.out.println("文件" + fileStatus.getPath().getName());

			}
		}
	}

	/**
	 * IO流上传
	 * 
	 * @throws Exception
	 */
	@Test
	public void upload() throws Exception {

		try (OutputStream out = fs.create(new Path("/mill/abc.mov")).getWrappedStream();
				InputStream in = Files.newInputStream(Paths.get("/Users/mill/aaa.mp4"), StandardOpenOption.READ);) {
			int len;
			byte[] b = new byte[4096];
			while ((len = in.read(b)) != -1) {
				out.write(b, 0, len);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * IO流下载
	 * 
	 * @throws Exception
	 */
	@Test
	public void downLoad() throws Exception {

		try (FSDataInputStream in = fs.open(new Path("/mill/abc.mov"));
				OutputStream out = Files.newOutputStream(Paths.get("/Users/mill/abc.mp4"),
						StandardOpenOption.CREATE_NEW);) {
			int len;
			byte[] b = new byte[4096];
			while ((len = in.read(b)) != -1) {
				out.write(b, 0, len);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 指定从跳过多少个字符
	 * @throws Exception
	 */
	@Test
	public void randomAccess() throws Exception {
		FSDataInputStream open = fs.open(new Path("/abcd.txt"));
		open.seek(6);
		open.seek(3);//第二次指定把第一次覆盖，最终跳过3个字节，也就是一个汉子
		try (BufferedReader read = new BufferedReader(new InputStreamReader(open));) {
			String str;
			while ((str = read.readLine()) != null) {
				System.out.println(str);
			}
		}
	}
	
	@Test
	public void append() throws Exception {
		FSDataOutputStream append = fs.append(new Path("/user/hive/warehouse/jone.db/freind/info.txt"));
		append.writeUTF("7,梁坤");
		append.close();
	}

	@After
	public void after() {
		try {
			if (fs != null)
				fs.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
